package com.soushin.mvpdemo.module.rxjava.example;

import com.blankj.ALog;
import com.google.gson.Gson;
import com.zhouyou.http.EasyHttp;
import com.zhouyou.http.callback.SimpleCallBack;
import com.zhouyou.http.exception.ApiException;
import com.zhouyou.http.subsciber.ProgressSubscriber;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.ObservableSource;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.functions.Function3;
import io.reactivex.schedulers.Schedulers;
import okhttp3.Call;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import com.soushin.mvpdemo.Configure;
import com.soushin.mvpdemo.Utils.AppUtils;
import com.soushin.mvpdemo.base.BasePresenter;
import com.soushin.mvpdemo.model.ResultTest1Model;
import com.soushin.mvpdemo.model.WeatherModel;
import com.soushin.mvpdemo.network.HttpClient;

/**
 * Created by SouShin on 2018/11/161611.
 */
public class ExamplePresenter extends BasePresenter<ExampleView> {
    private Disposable mDisposable;
    @Override
    public void subscribe(ExampleView mvpView) {
        super.subscribe(mvpView);
    }

    public void doSomeThing() {
        getMvpView().setResult("");
        switch (getMvpView().getClickPosition()) {
            case 0:
                goThreadRequest();
                break;
            case 1:
                goSingleNetWork();
                break;
            case 2:
                getWeatherInfo();
                break;
            case 3:
                goZipRequest();
                break;
            case 4:
                goNestedRequest();
                break;
            case 5:
                goConcatRequest();
                break;
            case 6:
                goDebounceRequest("1213");
                break;
            case 7:
                goPollRequest();
                break;
            case 8:
                onMergeRequest();
                break;
        }
    }

    /**
     * 单一的网络请求
     */
    private void goSingleNetWork() {

        mDisposable = Observable.create(new ObservableOnSubscribe<Response>() {
            @Override
            public void subscribe(ObservableEmitter<Response> e) throws Exception {
                Request.Builder builder = new Request.Builder()
                        .url(Configure.BASE_URL + "101120201.html")
                        .get();
                Request request = builder.build();
                Call call = new OkHttpClient().newCall(request);
                Response response = call.execute();
                e.onNext(response);
            }
        }).map(new Function<Response, ResultTest1Model>() {
            @Override
            public ResultTest1Model apply(Response response) throws Exception {
                ALog.e("map 线程", Thread.currentThread().getName());
                if (response.isSuccessful() && response.body() != null) {
                    String content = response.body().string();
                    ALog.e("map:转换前", content);
                    return new Gson().fromJson(content, ResultTest1Model.class);
                }
                return null;
            }
        }).observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<ResultTest1Model>() {
                    @Override
                    public void accept(ResultTest1Model weatherModelResultModel) throws Exception {
                        ALog.e("doOnNext 线程:", Thread.currentThread().getName());
                        getMvpView().getResultTextView().append("\ndoOnNext 线程:" + Thread.currentThread().getName() + "\n");
                        ALog.e("doOnNext: 保存成功", weatherModelResultModel.getWeatherinfo().toString());
                        getMvpView().getResultTextView().append("doOnNext: 保存成功：" + weatherModelResultModel.getWeatherinfo().toString() + "\n");
                    }
                }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<ResultTest1Model>() {
                    @Override
                    public void accept(ResultTest1Model weatherModelResultModel) throws Exception {
                        ALog.e("subscribe 线程", Thread.currentThread().getName());
                        getMvpView().getResultTextView().append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");
                        ALog.e("成功", weatherModelResultModel.getWeatherinfo().toString());
                        getMvpView().getResultTextView().append("成功:" + weatherModelResultModel.getWeatherinfo().toString() + "\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        ALog.e("subscribe 线程", Thread.currentThread().getName());
                        getMvpView().getResultTextView().append("\nsubscribe 线程:" + Thread.currentThread().getName() + "\n");

                        ALog.e("失败：" + throwable.toString());
                        getMvpView().getResultTextView().append("失败：" + throwable.getMessage() + "\n");
                    }
                });
    }

    /**
     * 自定义的网络请求
     * 青岛
     */
    private void getWeatherInfo() {
        HttpClient.getWeatherInfo(101120201, new SimpleCallBack<WeatherModel>() {
            @Override
            public void onError(ApiException e) {
                getMvpView().showToasty(AppUtils.getError(e.getCode(), e.getMessage()));
            }

            @Override
            public void onSuccess(WeatherModel weatherModel) {
                getMvpView().setResult(weatherModel.toString());
            }
        });
    }

    /**
     * zip 操作符的使用场景
     * 结合多个接口的数据再更新 UI
     */
    private void goZipRequest() {
        Observable<WeatherModel> qdObservable = HttpClient.getObservable(101120201);
        Observable<WeatherModel> bjObservable = HttpClient.getObservable(101010100);
        Observable<WeatherModel> shObservable = HttpClient.getObservable(101030100);
        Observable.zip(qdObservable, bjObservable, shObservable, new Function3<WeatherModel, WeatherModel, WeatherModel, List<WeatherModel>>() {
            @Override
            public List<WeatherModel> apply(WeatherModel weatherModel, WeatherModel weatherModel2, WeatherModel weatherModel3) throws Exception {
                List<WeatherModel> weatherList = new ArrayList<>();
                weatherList.add(weatherModel);
                weatherList.add(weatherModel2);
                weatherList.add(weatherModel3);
                ALog.e("zip线程",Thread.currentThread().getName());
                return weatherList;
            }
        }).subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()).subscribe(new ProgressSubscriber<List<WeatherModel>>(getMvpView().getContext()) {
                    @Override
                    public void onError(ApiException e) {
                        super.onError(e);
                        getMvpView().setResult(AppUtils.getError(e.getCode(), e.getMessage()));
                    }

                    @Override
                    public void onNext(List<WeatherModel> weatherModels) {
                        super.onNext(weatherModels);
                        ALog.e("subscribe线程",Thread.currentThread().getName());
                        String msg = "青岛天气:" + weatherModels.get(0).toString()
                                + "\n\n北京天气:" + weatherModels.get(1).toString()
                                + "\n\n天津天气:" + weatherModels.get(1).toString();
                        getMvpView().setResult(msg);
                    }
                });
    }


    /**
     * 嵌套请求
     * 一个接口的请求依赖另一个API请求返回的数据
     * 例如：先登录获取token后，再去请求用户信息接口
     */
    private void goNestedRequest() {
        Observable<WeatherModel> observable = HttpClient.getObservable(101120201);
        observable.flatMap(new Function<WeatherModel, ObservableSource<WeatherModel>>() {
            @Override
            public ObservableSource<WeatherModel> apply(WeatherModel weatherModel) throws Exception {
                ALog.e("flatMap 线程:", Thread.currentThread().getName());
                ALog.e("flatMap 成功", weatherModel.toString());
                //这个地方只是举例 所以没有取weatherModel里面的数据  实际操作第二个接口里的参数是要在第一个参数里面取值的
                return HttpClient.getObservable(101010100);
            }
        }).subscribe(new ProgressSubscriber<WeatherModel>(getMvpView().getContext()) {
            @Override
            public void onError(ApiException e) {
                super.onError(e);
                getMvpView().setResult(AppUtils.getError(e.getCode(), e.getMessage()));
            }

            @Override
            public void onNext(WeatherModel weatherModel) {
                super.onNext(weatherModel);
                ALog.e("onNext线程", Thread.currentThread().getName());
                getMvpView().setResult("北京天气:" + weatherModel.toString());
            }
        });
    }

    /**
     * Concat 先读取缓存数据并展示UI再获取网络数据刷新UI
     * <p>
     * 1、concat 可以做到不交错的发射两个甚至多个 Observable 的发射物;
     * 2、并且只有前一个 Observable 终止（onComplete）才会订阅下一个 Observable
     */
    private boolean isFromNet = false;

    private void goConcatRequest() {

        Observable<WeatherModel> cache = Observable.create(new ObservableOnSubscribe<WeatherModel>() {
            @Override
            public void subscribe(ObservableEmitter<WeatherModel> e) throws Exception {
                ALog.e("create当前线程", Thread.currentThread().getName());
                WeatherModel weatherModel = new WeatherModel();
                // 在操作符 concat 中，只有调用 onComplete 之后才会执行下一个 Observable
                if (weatherModel != null) { // 如果缓存数据不为空，则直接读取缓存数据，而不读取网络数据
                    isFromNet = false;
                    ALog.e("subscribe: 读取缓存数据:");
                    e.onNext(weatherModel);
                } else {
                    isFromNet = true;
                    ALog.e("subscribe: 读取网络数据:");
                    e.onComplete();
                }
            }
        });
        Observable<WeatherModel> network = HttpClient.getObservable(101120201);

         mDisposable = Observable.concat(cache, network)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<WeatherModel>() {
                    @Override
                    public void accept(WeatherModel weatherModel) throws Exception {
                        ALog.e("subscribe 成功:" + Thread.currentThread().getName());
                        if (isFromNet) {
                            getMvpView().getResultTextView().append("accept : 网络获取数据设置缓存: \n");
                            ALog.e("accept : 网络获取数据设置缓存:" + weatherModel.toString());
                        }
                        getMvpView().getResultTextView().append("accept: 读取数据成功:" + weatherModel.toString() + "\n");
                        ALog.e("accept: 读取数据成功:" + weatherModel.toString());
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        ALog.e("subscribe 失败", Thread.currentThread().getName());
                        ALog.e("accept: 读取数据失败", throwable.getMessage());
                        getMvpView().getResultTextView().append("accept: 读取数据失败：" + throwable.getMessage() + "\n");
                    }
                });

        /*EasyHttp.get("101120201.html")
                .readTimeOut(30 * 1000)//测试局部读超时30s
                .cacheMode(CacheMode.DEFAULT)
                .cacheKey(WeatherModel.class.getSimpleName())//缓存key
                .retryCount(5)//重试次数
                .cacheTime(5 * 60)//缓存时间300s，默认-1永久缓存  okhttp和自定义缓存都起作用
                //.okCache(new Cache());//okhttp缓存，模式为默认模式（CacheMode.DEFAULT）才生效
                //.cacheDiskConverter(new GsonDiskConverter())//默认使用的是 new SerializableDiskConverter();
                .cacheDiskConverter(new SerializableDiskConverter())//默认使用的是 new SerializableDiskConverter();
                .execute(new SimpleCallBack<CacheResult<WeatherModel>>() {

                    @Override
                    public void onError(ApiException e) {
                        //请求失败
                    }

                    @Override
                    public void onSuccess(CacheResult<WeatherModel> cacheResult) {
                        String from = "";
                        if (cacheResult.isFromCache) {
                            from = "我来自缓存";
                        } else {
                            from = "我来自远程网络";
                        }
                    }
                });*/
    }

    /**
     * debounce 操作符可以过滤掉发射频率过快的数据项
     * 设想情景：
     * 1、输入框数据变化时就要进行网络请求，这样会产生大量的网络请求。这时候可以通过debounce进行处理
     * 2、点击一次按钮就进行一次网络请求
     */
    private void goDebounceRequest(String key) {
        mDisposable = Observable.just(key)
                .debounce(2, TimeUnit.SECONDS)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        ALog.e("subscribe 成功:" + Thread.currentThread().getName());
                        getWeatherInfo();
                    }
                });
    }

    /**
     * 在 RxJava 2.x 中，采用 interval 操作符实现间隔任务
     * 使用场景:长连接任务时给服务端发送心跳包
     * filter 过滤操作符
     */
    private void goPollRequest() {
        mDisposable = Flowable.interval(1, TimeUnit.SECONDS)
                .doOnNext(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        ALog.e("accept: doOnNext  ", aLong);
                    }
                }).subscribeOn(Schedulers.computation())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(@NonNull Long aLong) throws Exception {
                        ALog.e("accept: 设置文本 ", aLong);
                        getMvpView().getResultTextView().append("accept: 设置文本 ：" + aLong + "\n");
                    }
                });
    }

    /**
     * 采用merge合并请求
     */
    private void onMergeRequest() {
        Observable<WeatherModel> qdObservable = HttpClient.getObservable(101120201);
        Observable<WeatherModel> bjObservable = HttpClient.getObservable(101010100);

        //merge和mergeDelayError都是合并，但是需要注意二者区别。
        //merge：合并的请求，如果有一个接口报错了，就立马报错，会终止整个流，另外的接口也不会请求。
        //mergeDelayError：合并的请求，如果有一个接口报错了，会延迟错误处理，后面的接口会继续执行没有被中断。
        //使用时根据需要选用merge/mergeDelayError
        mDisposable=Observable.merge(qdObservable,bjObservable)
                .subscribe(new Consumer<WeatherModel>() {
                    @Override
                    public void accept(WeatherModel weatherModel) throws Exception {
                        //如果两个接口返回的对象不一样,可以用Object接收，因为两个接口请求返回的数据类型不是一样的，如果是一样的就用具体的对象接收就可以了，
                        // 需要instanceof的判断,其实可以加一个flatmap操作符 就不用加判断了
                        getMvpView().getResultTextView().append(weatherModel.toString()+"\n");
                    }
                }, new Consumer<Throwable>() {
                    @Override
                    public void accept(Throwable throwable) throws Exception {
                        getMvpView().showToasty(throwable.getMessage());
                    }
                });
    }

    /**
     * 线程调度
     */
    private void goThreadRequest() {

        mDisposable=Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                ALog.e( "Observable 线程是" + Thread.currentThread().getName());
                e.onNext(1);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        ALog.e( "observeOn切换到主线程后，当前线程是" + Thread.currentThread().getName());

                    }
                })
                .observeOn(Schedulers.io())
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(@NonNull Integer integer) throws Exception {
                        ALog.e( "observeOn切换到io线程后，当前线程是" + Thread.currentThread().getName());
                    }
                });
    }



    @Override
    public void unsubscribe() {
        super.unsubscribe();
        EasyHttp.cancelSubscription(mDisposable);
    }
}
